package com.example.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

public class FluxDemo3 {

    public static void main(String[] args) {

        Flux<String> flux = Flux.range(1, 9)
                /**
                 * 挂载点
                 */
                .map(
                        /**
                         * 回调函数
                         */
                        integer -> "数据：" + integer
                )
                .doOnNext(System.out::println)
                .doOnComplete(() -> {
                    System.out.println("数据流完成");
                })
                .doFinally(signalType -> {
                    System.out.println("完成：" + signalType);
                });

        flux.subscribe(new BaseSubscriber<String>() {
            /**
             * 钩子函数
             */
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("钩子函数，订阅关系" + subscription);
                request(1);
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("钩子函数，接收数据：" + value);
                request(1);
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("钩子函数，数据流完成");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("钩子函数，数据流异常：" + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("钩子函数，数据流取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("钩子函数，数据流完成：" + type);
            }
        });

    }

}
